package com.lili.collection.core.runner.pipe;

import com.lili.collection.core.context.Context;
import com.lili.collection.core.context.PipeContext;
import com.lili.collection.core.context.StageContext;
import com.lili.collection.core.context.impl.GenericPipeContext;
import com.lili.collection.core.context.impl.SingleValueExchangeStageContext;
import com.lili.collection.core.error.InputRunnerException;
import com.lili.collection.core.error.OutputRunnerException;
import com.lili.collection.core.error.PluginRuntimeException;
import com.lili.collection.core.event.StageEvent;
import com.lili.collection.core.handler.IExecValidateHandler;
import com.lili.collection.core.runner.StageRunner;
import com.lili.collection.core.runner.handler.StageContextChangedHandlerFactory;
import com.lili.collection.core.runner.result.ErrOutputData;
import com.lili.collection.core.runner.result.RunnerResult;
import com.lili.collection.core.runner.stage.GenericInputRunner;
import com.lili.collection.core.runner.stage.GenericOutputRunner;
import lombok.Getter;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.util.HashMap;
import java.util.Map;

/**
 * 通用管道Runner
 *
 * @author lili
 * @date 2023/3/7 11:34
 */
@Slf4j
public class SyncPipeRunner extends AbstractPipeRunner {

    @Getter
    private PipeContext pipeContext;

    private Throwable runThrowable=null;

    public SyncPipeRunner(Context context) {
        this.pipeContext = new GenericPipeContext(context, new SingleValueExchangeStageContext());
    }

    @Override
    public void run() {
        long start = System.currentTimeMillis();
        try {
            ensureStageRunner();
            if (this.stageRunnerList.size() == 0) {
                return;
            }
            if(this.runThrowable!=null){
                throw this.runThrowable;
            }
            //前置处理
            this.execPreHandle();
            StageRunner.StageType currentStageType = null;
            do {
                for (StageRunner stageRunner : this.stageRunnerList) {
                    checkStopSignal();
                    stageRunner.run();
                    currentStageType = stageRunner.getStageType();
                    checkStageRecordState(currentStageType);
                }
            } while (checkStageContext(currentStageType));
        } catch (Throwable throwable) {
            runThrowable=throwable;
            boolean noSetErrorMsg= true;
            for (StageRunner runner : this.stageRunnerList) {
                RunnerResult runnerResult = runner.getRunnerResult();
                if(runnerResult!=null){
                    runnerResult.setSuccessed(false);
                    //目标支持事务则清空采集记录
                    ErrOutputData errOutputData = runnerResult.getErrOutputData();
                    if((throwable instanceof OutputRunnerException) && (runner instanceof GenericOutputRunner)){
                        noSetErrorMsg=false;
                        if(StringUtils.isBlank(errOutputData.getErrorMessage())){
                            errOutputData.setErrorMessage(throwable.toString());
                        }
                    }
                    if((throwable instanceof InputRunnerException) && (runner instanceof GenericInputRunner)){
                        noSetErrorMsg=false;
                        if(StringUtils.isBlank(errOutputData.getErrorMessage())){
                            errOutputData.setErrorMessage(throwable.toString());
                        }
                    }
                }
            }
            if(noSetErrorMsg){
                this.fillThrowableMessageToRunnerResult(throwable);
            }
            this.clearTxCount();
            log.error("同步管道异常!", throwable);
        } finally {
            RunnerResult outRunnerResult = this.getOutputRunner().getRunnerResult();
            if(outRunnerResult!=null){
                int deleteOffsetNum = outRunnerResult.getErrOutputData().getDeleteOffsetNum();
                RunnerResult inputRunnerResult = this.getInputRunner().getRunnerResult();
                if(deleteOffsetNum>0 && inputRunnerResult!=null){
                    outRunnerResult.decrementCount(deleteOffsetNum);
                    inputRunnerResult.decrementCount(deleteOffsetNum);
                }
            }
            try {
                dispose();
            } catch (Exception e) {
                log.error("关闭异常！",e);
            }
            long end = System.currentTimeMillis();
            log.info("-->完成同步，耗时：{} ms", end - start);
        }
    }

    private void clearTxCount() {
        try {
            boolean enableTransaction= getPipeContext().getContext().getOutputConfig().isEnableTransaction();
            RunnerResult outRunnerResult= this.getOutputRunner().getRunnerResult();
            if(outRunnerResult!=null && enableTransaction){
                outRunnerResult.addErrorCount(outRunnerResult.getSuccessRecordCount());
                outRunnerResult.setSuccessRecordCount(0);
            }
        } catch (Exception e) {
            log.error("清除事务记录异常！",e);
        }
    }

    private void checkStopSignal() {
        Context context = this.pipeContext.getContext();
        if(context.getStopSignal()){
            throw new PluginRuntimeException("采集强制停止异常！");
        }
        IExecValidateHandler validateHandler = context.getValidateHandler();
        if(validateHandler!=null){
            if(!validateHandler.onValidate()){
                throw new PluginRuntimeException("任务执行校验失败！");
            }
        }
    }

    private void fillThrowableMessageToRunnerResult(Throwable throwable) {
        this.stageRunnerList.forEach((runner)->{
            RunnerResult runnerResult = runner.getRunnerResult();
            if(runnerResult!=null){
                ErrOutputData errOutputData = runnerResult.getErrOutputData();
                if(errOutputData!=null && StringUtils.isBlank(errOutputData.getErrorMessage())){
                    errOutputData.setErrorMessage(throwable.toString());
                }
            }
        });
    }

    @Override
    public String getOutputOffset() {
        try {
            this.ensureStageRunner();
            return this.getOutputRunner().queryInitialOffset();
        } catch (Throwable throwable) {
            runThrowable=throwable;
            log.error("获取偏移量错误!",throwable);
        }
        return null;
    }

    @Override
    public Map<String, RunnerResult> getPipeRunnerResult() {
        Map<String, RunnerResult> resultMap=new HashMap<>();
        for (StageRunner stageRunner : this.stageRunnerList) {
            resultMap.put(stageRunner.getClass().getName(),stageRunner.getRunnerResult());
        }
        return resultMap;
    }

    private void handleOutput() {
        //强行消费Output交换数据
        StageContext stageContext = getPipeContext().getExchangeStageContext().take(StageRunner.StageType.Output);
        if (stageContext == null)
            return;
    }

    @Override
    public void onStageEventTrigger(StageEvent stageEvent, StageRunner.StageType stageType, Object params) {
        StageContextChangedHandlerFactory.createHandler(stageType).handle(this, stageEvent, params);
    }

    @SneakyThrows
    private void checkStageRecordState(StageRunner.StageType stageType) {
        StageContext stageContext = getPipeContext().getExchangeStageContext().peek(stageType);
        if (stageContext != null && stageContext.getStageRecordState() == StageContext.StageRecordState.ALL_ERROR) {
            if(stageType == StageRunner.StageType.Output){
                throw new OutputRunnerException(stageContext.getException());
            }else if(stageType == StageRunner.StageType.Input){
                throw new InputRunnerException(stageContext.getException());
            }else {
                throw new PluginRuntimeException(stageContext.getException());
            }
        }
    }

    protected boolean checkStageContext(StageRunner.StageType stageType) {
        if (stageType == null) return false;
        StageContext stageContext = getPipeContext().getExchangeStageContext().peek(stageType);
        if (stageType == StageRunner.StageType.Output)
            handleOutput();
        if (stageContext.getStageContextState() == StageContext.StageContextState.COMPLETE)
            return false;
        return true;
    }
}